package com.group2.edu.realtime.dws;/**
 * @Author: Wxs
 * @Description: TODO
 * @Date: 2024/12/16 15:05
 */

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.group2.edu.realtime.common.base.BaseApp;
import com.group2.edu.realtime.common.bean.InteractionCourseReviewWindowBean;
import com.group2.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.group2.edu.realtime.common.function.DimAsyncFunction;
import com.group2.edu.realtime.common.util.DateFormatUtil;
import com.group2.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;

/**
 * @Author: Wxs
 * @Description: TODO
 * @Date: 2024/12/16 15:05
 * 需要启动的进程
 *       zk、kafka、maxwell、redis、hdfs、hbase、doris、DwdBaseTrain、DwsInteractionCourseReviewWindow
 */
public class DwsInteractionCourseReviewWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsInteractionCourseReviewWindow().start(
                10033,
                4,
                "dws_interaction_course_review_window",
                "dwd_interaction_review"
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        //TODO 对流中数据类型进行转换
        SingleOutputStreamOperator<InteractionCourseReviewWindowBean> beanDS = kafkaStrDS.flatMap(
                new FlatMapFunction<String, InteractionCourseReviewWindowBean>() {
                    @Override
                    public void flatMap(String value, Collector<InteractionCourseReviewWindowBean> out) throws Exception {
                        JSONObject jsonObj = JSON.parseObject(value);
                        String courseId = jsonObj.getString("course_id");
                        Long reviewStars = jsonObj.getLong("review_stars");
                        Long ts = jsonObj.getLong("ts") * 1000;
                        out.collect(InteractionCourseReviewWindowBean.builder()
                                .courseId(courseId)
                                .reviewTotalStars(reviewStars)
                                .reviewUserCount(1l)
                                .goodReviewUserCount(reviewStars == 5l ? 1l : 0l)
                                .ts(ts)
                                .build());
                    }
                }
        );
        //TODO 指定水位线生成策略以及事件时间字段
        SingleOutputStreamOperator<InteractionCourseReviewWindowBean> withWatermarkDS = beanDS.assignTimestampsAndWatermarks(WatermarkStrategy.<InteractionCourseReviewWindowBean>forMonotonousTimestamps()
                .withTimestampAssigner(
                        new SerializableTimestampAssigner<InteractionCourseReviewWindowBean>() {
                            @Override
                            public long extractTimestamp(InteractionCourseReviewWindowBean element, long recordTimestamp) {
                                return element.getTs();
                            }
                        }
                ));
        //TODO 按照课程id分组
        KeyedStream<InteractionCourseReviewWindowBean, String> keyedDS = withWatermarkDS.keyBy(
                new KeySelector<InteractionCourseReviewWindowBean, String>() {
                    @Override
                    public String getKey(InteractionCourseReviewWindowBean value) throws Exception {
                        return value.getCourseId();
                    }
                }
        );
        //TODO 开窗聚合
        SingleOutputStreamOperator<InteractionCourseReviewWindowBean> reduceDS = keyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10l)))
                .reduce(
                        new ReduceFunction<InteractionCourseReviewWindowBean>() {
                            @Override
                            public InteractionCourseReviewWindowBean reduce(InteractionCourseReviewWindowBean value1, InteractionCourseReviewWindowBean value2) throws Exception {
                                value1.setReviewTotalStars(value1.getReviewTotalStars() + value2.getReviewTotalStars());
                                value1.setReviewUserCount(value1.getReviewUserCount() + value2.getReviewUserCount());
                                value1.setGoodReviewUserCount(value1.getGoodReviewUserCount() + value2.getGoodReviewUserCount());
                                return value1;

                            }
                        }, new ProcessWindowFunction<InteractionCourseReviewWindowBean, InteractionCourseReviewWindowBean, String, TimeWindow>() {
                            @Override
                            public void process(String s, ProcessWindowFunction<InteractionCourseReviewWindowBean, InteractionCourseReviewWindowBean, String, TimeWindow>.Context context, Iterable<InteractionCourseReviewWindowBean> elements, Collector<InteractionCourseReviewWindowBean> out) throws Exception {
                                String stt = DateFormatUtil.tsToDateTime(context.window().getStart());
                                String edt = DateFormatUtil.tsToDateTime(context.window().getEnd());
                                String CurDate = DateFormatUtil.tsToDate(context.window().getStart());
                                for (InteractionCourseReviewWindowBean element : elements) {
                                    element.setStt(stt);
                                    element.setEdt(edt);
                                    element.setCurDate(CurDate);
                                    element.setTs(System.currentTimeMillis());
                                    out.collect(element);
                                }

                            }
                        }
                );
        //reduceDS.print();
        //TODO 维度关联课程信息
        SingleOutputStreamOperator<InteractionCourseReviewWindowBean> withCoursenameDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<InteractionCourseReviewWindowBean>() {
                    @Override
                    public void addDims(InteractionCourseReviewWindowBean obj, JSONObject dimJsonObj) {
                        obj.setCourseName(dimJsonObj.getString("course_name"));
                    }

                    @Override
                    public String getTableName() {
                        return "dim_course_info";
                    }

                    @Override
                    public String getRowKey(InteractionCourseReviewWindowBean obj) {
                        return obj.getCourseId();
                    }
                }, 60, TimeUnit.SECONDS
        );
        withCoursenameDS.print();
        withCoursenameDS.map(new BeanToJsonStrMapFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink("dws_interaction_course_review_window"));

    }
}
